package com.hilin.etcd.discovery.handler;


import com.hilin.common.util.EncryptUtils;
import com.hilin.common.util.EntityUtils;
import com.hilin.common.util.IpUtils;
import com.hilin.etcd.discovery.config.HilinEtcdDiscoveryConfig;
import com.hilin.etcd.discovery.entity.HilinEtcdRegisterInfo;
import lombok.extern.slf4j.Slf4j;
import mousio.etcd4j.EtcdClient;
import mousio.etcd4j.responses.EtcdKeysResponse;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.util.ObjectUtils;

import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

@Slf4j
public class HilinEtcdRegisterHandler {

    private static boolean isInit = true;

    /**
     * 服务注册的根目录
     */
    private static final String REG_ROOT_KEY = "/com/hilin/register/";

    /**
     * 当前服务注册目录的父目录
     * 该目录包含了同一serverId的所有服务，因此是一个列表
     */
    private static final String REG_SERVER_PARENT_KEY = REG_ROOT_KEY + "%s/";

    /**
     * 服务信息在etcd中的key
     * 第一个%s: 服务ID
     * 第二个%s: MD5(ip:端口)
     */
    private static final String REG_KEY = REG_SERVER_PARENT_KEY + "%s";

    private final EtcdClient etcdClient;
    private Map<String, List<HilinEtcdRegisterInfo>> servers;

    @Value("${server.port}")
    private int port;

    /**
     * 心跳发送线程
     */
    private static Thread HEART_THREAD = null;

    /**
     * 服务监控线程
     */
    private static Thread SERVER_THREAD = null;

    public HilinEtcdRegisterHandler(HilinEtcdDiscoveryConfig etcdDiscoveryConfig, EtcdClient etcdClient) {
        this.etcdClient = etcdClient;
        String ip = IpUtils.getLocalIp();
        final String regKey = String.format(REG_KEY, etcdDiscoveryConfig.getServerId(), EncryptUtils.getMd5(ip + ":" + port));
        if (HEART_THREAD == null) {
            HEART_THREAD = new Thread(() -> {
                while (true) {
                    try {
                        if (port == 0) {
                            Thread.sleep(1000L);
                            continue;
                        }
                        if (!isInit) {
                            Thread.sleep(etcdDiscoveryConfig.getHeartRate() * 1000L);
                        }
                        isInit = false;
                        String info = EntityUtils.toJsonString(HilinEtcdRegisterInfo.builder()
                                .ip(ip)
                                .port(port)
                                .serverId(etcdDiscoveryConfig.getServerId())
                                .lastSendHeartTime(System.currentTimeMillis())
                                .build());
                        etcdClient.put(regKey, info).ttl(etcdDiscoveryConfig.getHeartTimeout()).send();
                        if (log.isDebugEnabled()) {
                            log.debug("服务注册成功 {}", info);
                        }
                    } catch (Exception e) {
                        log.error("心跳注册失败  " + EntityUtils.toJsonString(etcdDiscoveryConfig), e);
                    }
                }
            });
            HEART_THREAD.start();
        }
        if (SERVER_THREAD == null) {
            SERVER_THREAD = new Thread(() -> {
                while (true) {
                    try {
                        freshServers();
                        Thread.sleep(etcdDiscoveryConfig.getHeartRate() * 1000L);
                    } catch (Exception e) {
                        log.error("心跳注册失败  " + EntityUtils.toJsonString(etcdDiscoveryConfig), e);
                    }
                }
            });
            SERVER_THREAD.start();
        }
    }

    private void freshServers() throws Exception {
        Map<String, List<HilinEtcdRegisterInfo>> servers = new ConcurrentHashMap<>();
        // 查询所有的服务
        List<EtcdKeysResponse.EtcdNode> serverList = etcdClient.get(REG_ROOT_KEY).send().get().node.nodes;
        if (!ObjectUtils.isEmpty(serverList)) {
            for (EtcdKeysResponse.EtcdNode etcdNode : serverList) {
                if (etcdNode.dir) {
                    EtcdKeysResponse serverNode = etcdClient.get(etcdNode.getKey() + "/").send().get();
                    if (ObjectUtils.isEmpty(serverNode))
                        continue;
                    if (ObjectUtils.isEmpty(serverNode.node) || ObjectUtils.isEmpty(serverNode.node.nodes)) {
                        continue;
                    }
                    for (EtcdKeysResponse.EtcdNode node : serverNode.node.nodes) {
                        HilinEtcdRegisterInfo serverInfo = EntityUtils.jsonStr2Entity(node.value, HilinEtcdRegisterInfo.class);
                        if (serverInfo != null && !ObjectUtils.isEmpty(serverInfo.getServerId())) {
                            servers.putIfAbsent(serverInfo.getServerId(), new ArrayList<>());
                            servers.get(serverInfo.getServerId()).add(serverInfo);
                            if (log.isDebugEnabled()) {
                                log.info("查询到服务 {} {}:{}", serverInfo.getServerId(), serverInfo.getIp(), serverInfo.getPort());
                            }
                        }
                    }

                }
            }
        }
        synchronized (REG_ROOT_KEY) {
            if (!ObjectUtils.isEmpty(this.servers)) {
                this.servers.clear();
            }
            this.servers = servers;
        }
    }

    public Map<String, List<HilinEtcdRegisterInfo>> getServers() {
        synchronized (REG_ROOT_KEY) {
            return this.servers;
        }
    }

    public List<HilinEtcdRegisterInfo> getServerByServerId(String serverId) {
        synchronized (REG_ROOT_KEY) {
            return this.servers.get(serverId);
        }
    }
}
